package com;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.File;
import java.io.IOException;
import java.util.Arrays;

/**CommonFriebds输出结果：
 * A	I,K,C,B,G,F,H,O,D,   谁的好友有A，即 I,K,C,B,G,F,H,O,D的共同好友是A
 * B	A,F,J,E,        谁的好友有B，    即A,F,J,E,的共同好友是B
 * C	A,E,B,H,F,G,K,
 * D	G,C,K,A,L,F,E,H,
 * E	G,M,L,H,A,F,B,D,
 * F	L,M,D,C,G,A,
 * G	M,
 * H	O,
 * I	O,C,
 * J	O,
 * K	B,
 * L	D,E,
 * M	E,F,
 * O	A,H,I,J,F,
 *
 * CommonFriends2输出结果：
 * A-B的共同好友是：	E C
 * A-C的共同好友是：	D F
 * A-D的共同好友是：	E F
 * A-E的共同好友是：	D B C
 * A-F的共同好友是：	O B C D E
 * A-G的共同好友是：	F E C D
 * A-H的共同好友是：	E C D O
 * A-I的共同好友是：	O
 * A-J的共同好友是：	O B
 * A-K的共同好友是：	D C
 * A-L的共同好友是：	F E D
 * A-M的共同好友是：	E F
 * B-C的共同好友是：	A
 * B-D的共同好友是：	A E
 * B-E的共同好友是：	C
 * B-F的共同好友是：	E A C
 * B-G的共同好友是：	C E A
 * B-H的共同好友是：	A E C
 * B-I的共同好友是：	A
 * B-K的共同好友是：	C A
 * B-L的共同好友是：	E
 * B-M的共同好友是：	E
 * B-O的共同好友是：	A
 * C-D的共同好友是：	A F
 * C-E的共同好友是：	D
 * C-F的共同好友是：	D A
 * C-G的共同好友是：	D F A
 * C-H的共同好友是：	D A
 * C-I的共同好友是：	A
 * C-K的共同好友是：	A D
 * C-L的共同好友是：	D F
 * C-M的共同好友是：	F
 * C-O的共同好友是：	I A
 * D-E的共同好友是：	L
 * D-F的共同好友是：	A E
 * D-G的共同好友是：	E A F
 * D-H的共同好友是：	A E
 * D-I的共同好友是：	A
 * D-K的共同好友是：	A
 * D-L的共同好友是：	E F
 * D-M的共同好友是：	F E
 * D-O的共同好友是：	A
 * E-F的共同好友是：	D M C B
 * E-G的共同好友是：	C D
 * E-H的共同好友是：	C D
 * E-J的共同好友是：	B
 * E-K的共同好友是：	C D
 * E-L的共同好友是：	D
 * F-G的共同好友是：	D C A E
 * F-H的共同好友是：	A D O E C
 * F-I的共同好友是：	O A
 * F-J的共同好友是：	B O
 * F-K的共同好友是：	D C A
 * F-L的共同好友是：	E D
 * F-M的共同好友是：	E
 * F-O的共同好友是：	A
 * G-H的共同好友是：	D C E A
 * G-I的共同好友是：	A
 * G-K的共同好友是：	D A C
 * G-L的共同好友是：	D F E
 * G-M的共同好友是：	E F
 * G-O的共同好友是：	A
 * H-I的共同好友是：	O A
 * H-J的共同好友是：	O
 * H-K的共同好友是：	A C D
 * H-L的共同好友是：	D E
 * H-M的共同好友是：	E
 * H-O的共同好友是：	A
 * I-J的共同好友是：	O
 * I-K的共同好友是：	A
 * I-O的共同好友是：	A
 * K-L的共同好友是：	D
 * K-O的共同好友是：	A
 * L-M的共同好友是：	E F
 *
 * @Auther: zhezhe
 * @Date: 2018/11/215:35
 * @Description:
 */
public class CommonFriends2 {
    public static class MapFriends2 extends Mapper<LongWritable, Text,Text,Text>{
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            //A	I,K,C,B,G,F,H,O,D,
            String line = value.toString();
            String[] split = line.split("\t");

            String[] split1 = split[1].split(",");

            Arrays.sort(split1);

            //该循环实现  例如，遍历第一行数据 A I,K,C,B,G,F,H,O,D, 时，i = 0，j = 1，split1[0]=I，split1[1]=K，I和K的共同好友是A
            for (int i = 0; i <split1.length-1 ; i++) {
                for (int j = i+1; j <split1.length; j++) {
                    //split1[i]和split1[j]的共同好友是split[0]
                    context.write(new Text(split1[i]+"-"+split1[j]),new Text(split[0]));
                }

            }
        }
    }

    public static class ReduceFriends2 extends Reducer<Text,Text,Text,Text>{
        @Override
        protected void reduce(Text key, Iterable<Text> values, Context context) throws IOException, InterruptedException {

            StringBuffer sbf = new StringBuffer();
            for (Text value : values) {
                sbf.append(value.toString()).append(" ");
            }
            //相同的key的value聚合，即 谁和谁的共同好友是谁谁谁
            context.write(new Text(key+"的共同好友是："),new Text(sbf.toString()));
        }
    }


    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(CommonFriends2.class);
        job.setMapperClass(CommonFriends2.MapFriends2.class);
        job.setReducerClass(CommonFriends2.ReduceFriends2.class);

        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(Text.class);

        job.setNumReduceTasks(1);

        File file = new File("E:\\Mapreduce案例\\CommonFriends\\数据\\output1");
        if (file.exists()){
            FileUtil.fullyDelete(file);
        }

        FileInputFormat.addInputPath(job,new Path("E:\\Mapreduce案例\\CommonFriends\\数据\\output\\*"));
        FileOutputFormat.setOutputPath(job,new Path("E:\\Mapreduce案例\\CommonFriends\\数据\\output1"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }
}
